package com.mall.order.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.OrderCoreService;
import com.mall.order.dto.CancelOrderRequest;
import com.mall.order.dto.CancelOrderResponse;
import com.mall.user.constants.SysRetCodeConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
@Slf4j
@Component
public class OrderConsumer {

    @Autowired
    OrderCoreService orderCoreService;


    private DefaultMQPushConsumer consumer;

    @Value("${mq.addr}")
    private String addr;
    @Value("${mq.topic}")
    private String topic;


    @PostConstruct
    public void init(){
        log.info("测试consumer！！！！");
        consumer = new DefaultMQPushConsumer("consumer_group");

        // 设置注册中心的地址
        consumer.setNamesrvAddr(addr);

        // 订阅topic
        try {
            consumer.subscribe(topic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            // 这个方法就是我们这个消息消费者收到了消息之后调用的方法
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {


                // 收到了延迟消息之后的逻辑
                MessageExt messageExt = msgs.get(0);
                byte[] body = messageExt.getBody();
                String bodyStr = new String(body);

                Map<String,Object> map  = JSON.parseObject(bodyStr,Map.class);

                String orderId = (String) map.get("orderId");

                CancelOrderRequest request = new CancelOrderRequest();
                request.setOrderId(orderId);

                CancelOrderResponse cancelOrderResponse = orderCoreService.orderCancle(request);

                if(cancelOrderResponse.getCode().equals(SysRetCodeConstants.SUCCESS.getCode())){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }else {
                    // 假如在这里消费失败的话，rocketmq默认会去重试，重试16次
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

            }
        });


        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        System.out.println("消息消费者启动成功！！！");


    }


}
